package simpledb.execution;

import simpledb.common.DbException;
import simpledb.common.Type;
import simpledb.storage.IntField;
import simpledb.storage.StringField;
import simpledb.storage.Tuple;
import simpledb.storage.TupleDesc;
import simpledb.transaction.TransactionAbortedException;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.util.*;

/**
 * Knows how to compute some aggregate over a set of IntFields.
 */
public class IntegerAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    /**
     * Aggregate constructor
     * 
     * @param gbfield
     *            the 0-based index of the group-by field in the tuple, or
     *            NO_GROUPING if there is no grouping
     * @param gbfieldtype
     *            the type of the group by field (e.g., Type.INT_TYPE), or null
     *            if there is no grouping
     * @param afield
     *            the 0-based index of the aggregate field in the tuple
     * @param what
     *            the aggregation operator
     */
    private int gbfield;
    private Type gbfiledtype;
    private int afield;
    private Op op;
    private Map<String, List<Integer>>gbIdxToAggList;

    public IntegerAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.gbfield=gbfield;
        this.afield=afield;
        this.gbfiledtype=gbfieldtype;
        this.op=what;
        gbIdxToAggList=new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the
     * constructor
     * 
     * @param tup
     *            the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        String key="NoGrouping";
        if(!(gbfield==Aggregator.NO_GROUPING)&&gbfiledtype==Type.INT_TYPE){
            IntField intField= (IntField) tup.getField(gbfield);
            key=Integer.toString(intField.getValue());
        }
        else if(!(gbfield==Aggregator.NO_GROUPING)){
            StringField stringField=(StringField) tup.getField(gbfield);
            key=stringField.getValue();
        }
        IntField intField= (IntField) tup.getField(afield);
        Integer val=intField.getValue();
        if(!gbIdxToAggList.containsKey(key)){
           gbIdxToAggList.put(key,new ArrayList<>());
        }
        gbIdxToAggList.get(key).add(val);

    }

    /**
     * Create a OpIterator over group aggregate results.
     * 
     * @return a OpIterator whose tuples are the pair (groupVal, aggregateVal)
     *         if using group, or a single (aggregateVal) if no grouping. The
     *         aggregateVal is determined by the type of aggregate specified in
     *         the constructor.
     */
    public OpIterator iterator() {
        // some code goes here
        return new OpIterator() {
            private Iterator iter = null;

            private int calculate(ArrayList<Integer> list) {
                assert !list.isEmpty() : "expect non empty list";
                int res = 0;
                switch (op) {
                    case AVG:
                        for (int i : list) {
                            res += i;
                        }
                        return res / list.size();
                    case MAX:
                        return Collections.max(list);
                    case MIN:
                        return Collections.min(list);
                    case SUM:
                        for (int i : list) {
                            res += i;
                        }
                        return res;
                    case COUNT:
                        return list.size();
                    default:
                        throw new NotImplementedException();
                }
            }

            @Override
            public void open() throws DbException, TransactionAbortedException {
                iter = gbIdxToAggList.entrySet().iterator();
            }

            @Override
            public boolean hasNext() throws DbException, TransactionAbortedException {
                if (iter == null)
                    throw new IllegalStateException("Operator not yet open");

                return iter.hasNext();
            }

            @Override
            public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
                if (iter == null)
                    throw new IllegalStateException("Operator not yet open");

                if (!iter.hasNext()) {
                    throw new NoSuchElementException();
                }

                /*HashMap.Entry pair = (HashMap.Entry) iter.next();
                ArrayList<Integer> l = (ArrayList<Integer>) pair.getValue();
                String key = (String) pair.getKey();
                IntField val = new IntField(calculate(l));*/

                //求它的next的话，首先需要把对应的列表取出来。
                Map.Entry<String,List<Integer>>pair= (Map.Entry<String, List<Integer>>) iter.next();
                ArrayList<Integer>list= (ArrayList<Integer>) pair.getValue();
                String key= pair.getKey();
                IntField val=new IntField(calculate(list));

                /*TupleDesc td = getTupleDesc();
                Tuple tp = new Tuple(td);*/

                TupleDesc td=getTupleDesc();
                Tuple tp=new Tuple(td);

                if (gbfield == Aggregator.NO_GROUPING) {
                    tp.setField(0, val);
                } else if (gbfiledtype == Type.INT_TYPE) {
                    IntField bgFieldValue = new IntField(Integer.parseInt(key));
                    tp.setField(0, bgFieldValue);
                    tp.setField(1, val);
                } else {
                    StringField bgFieldValue = new StringField(key, key.length());
                    tp.setField(0, bgFieldValue);
                    tp.setField(1, val);
                }

                return tp;
            }

            @Override
            public void rewind() throws DbException, TransactionAbortedException {
                close();
                open();
            }

            @Override
            public TupleDesc getTupleDesc() {
                if (gbfield == Aggregator.NO_GROUPING) {
                    return new TupleDesc(new Type[]{Type.INT_TYPE});
                }
                return new TupleDesc(new Type[]{gbfiledtype, Type.INT_TYPE});
            }

            @Override
            public void close() {
                iter = null;
            }
        };
    }

}
